package com.zc.websocket;

import com.alibaba.fastjson.JSON;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
import com.google.gson.reflect.TypeToken;
import com.ruoyi.common.utils.StringUtils;
import com.zc.common.constant.RedisStreamConstants;
import com.zc.common.core.redis.stream.RedisStream;
import com.zc.common.core.websocket.WebSocketService;
import com.zc.component.gateway.external.Message;
import com.zc.component.gateway.external.MessagingManager;
import com.zc.component.gateway.external.SubscribeRequest;
import com.zc.web.logger.ReactiveLogger;
import com.zc.websocket.constant.WebSocketMsgEvent;
import com.zc.websocket.dto.param.EventParam;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import reactor.core.Disposable;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.lang.reflect.Type;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;

/**
 * webSocket 事件监听器
 * 处理接收 Redis 缓存订阅消息
 * @author Athena-xiepufeng
 */
@Component
public class WebSocketStreamEventListener implements StreamListener<String, ObjectRecord<String, String>>
{
    private static final Logger log = LoggerFactory.getLogger(WebSocketStreamEventListener.class);

    @Resource
    private RedisStream redisStream;

    @Resource
    private MessagingManager messagingManager;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {

        String streamKay = message.getStream();
        RecordId recordId = message.getId();

        String key = RedisStreamConstants.WebSocketStreamEvent.KEY;
        String group = RedisStreamConstants.WebSocketStreamEvent.GROUP;

        if (!Objects.equals(streamKay, key)) {
            log.error("队列不匹配 接收 key: {}, 定义的 key：{}", streamKay, key);
            return;
        }
        
        try {

            Type type = new TypeToken<EventParam<JsonObject>>()
            {
            }.getType();

            Gson gson = new Gson();

            EventParam<JsonObject> eventParam =  gson.fromJson(message.getValue(), type);

            String subEvent = eventParam.getSubEvent();

            switch (subEvent) {
                // 订阅事件
                case WebSocketMsgEvent.SUB: {
                    SubscribeRequest subscribeRequest = gson.fromJson(eventParam.getContent(), new TypeToken<SubscribeRequest>() {}.getType());
                    if (subscribeRequest != null) {
                        subscribeRequest.setTokenSN(eventParam.getTokenSN());
                        handleSub(subscribeRequest);
                    }
                    break;
                }
                // 取消订阅事件
                case WebSocketMsgEvent.UNSUB: {
                    SubscribeRequest subscribeRequest = gson.fromJson(eventParam.getContent(), new TypeToken<SubscribeRequest>() {}.getType());
                    if (subscribeRequest != null) {
                        subscribeRequest.setTokenSN(eventParam.getTokenSN());
                        handleUnsub(subscribeRequest);
                    }
                    break;
                }
                default: {
                    throw new IllegalStateException("未知的 webSocket 事件类型: " + subEvent);
                }
            }

            // 消费完成后确认消费（ACK）
            redisStream.ack(streamKay, group, String.valueOf(recordId));
            // 消费完后直接删除消息
            redisStream.del(streamKay, String.valueOf(recordId));
        }catch (Exception e) {
            e.printStackTrace();
            log.error("消费异常 key: {}, 群组：{}, 消息Id：{}", streamKay, group, recordId);
        }
    }

    Map<String, Disposable> subs = new ConcurrentHashMap<>();

    /**
     * 订阅消息处理
     * @param subscribeRequest 订阅数据
     */
    public void handleSub(SubscribeRequest subscribeRequest) {

        if (subscribeRequest == null) {
            return;
        }

        String disposableId = subscribeRequest.getTokenSN() + subscribeRequest.getId();

        //重复订阅
        Disposable old = subs.get(disposableId);
        if (old != null && !old.isDisposed()) {
            return;
        }

        Disposable sub = messagingManager
                .subscribe(subscribeRequest)
                .doOnEach(ReactiveLogger.onError(err -> log.error("{}", err.getMessage(), err)))
                .onErrorResume(err -> Mono.just(Message.error(subscribeRequest.getId(), subscribeRequest.getTopic(), err)))
                .map(JSON::toJSONString)
                .doOnComplete(() -> {
                    log.debug("complete subscription:{}", subscribeRequest.getTopic());
                    subs.remove(disposableId);
                    Mono.just(JSON.toJSONString(Message.complete(subscribeRequest.getId())))
                            .as(msg -> Mono.defer(() -> {
                                WebSocketService.postEvent(subscribeRequest.getTokenSN(), WebSocketMsgEvent.SUB, msg);
                                return Mono.just(true);
                            }))
                            .subscribe();
                })
                .doOnCancel(() -> {
                    log.debug("cancel subscription:{}", subscribeRequest.getTopic());
                    subs.remove(disposableId);
                })
                .doOnNext(msg -> WebSocketService.postEvent(subscribeRequest.getTokenSN(), WebSocketMsgEvent.SUB, msg))
                .subscribe();

        if (!sub.isDisposed()) {
            subs.put(disposableId, sub);
        }
    }

    /**
     * 取消订阅处理
     * @param subscribeRequest 取消订阅数据
     */
    public void handleUnsub(SubscribeRequest subscribeRequest) {

        if (subscribeRequest == null) {
            return;
        }

        // 根据tokenSN清除所以订阅
        if(StringUtils.isEmpty(subscribeRequest.getId())) {
            subs.forEach((id, dispose) -> {
                if (id.startsWith(subscribeRequest.getTokenSN())) {
                    Optional.ofNullable(subs.remove(id))
                            .ifPresent(Disposable::dispose);
                }
            });
        }

        Optional.ofNullable(subs.remove(subscribeRequest.getTokenSN() + subscribeRequest.getId()))
                .ifPresent(Disposable::dispose);
    }
}
